package com.shujia.wyh.jinjie;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.List;

/*
    合理设计rowkey，通过手机号查询用户最近10条位置记录

    47BE1E866CFC071DB19D5E1C056BE28AE24C16E7
 */
public class DianXinDataDemo {
    private static Connection conn;
    private static Admin admin;
    public static void main(String[] args) throws Exception {
        init();
        //创建表描述器
        TableName dianxinTb1 = TableName.valueOf("dianxin_tb1");
//        createTable(dianxinTb1);

//        putDatas(dianxinTb1);
//        getData(dianxinTb1);

        getData2(dianxinTb1);






    }

    /*
        获取用户某一天在一个城市中的所有位置(组合过滤器)
        47BE1E866CFC071DB19D5E1C056BE28AE24C16E7
        20180503
        83211
     */
    public static void getData2(TableName tableName) throws Exception{
        //获取表的实例
        Table table = conn.getTable(tableName);

        //创建Scan的对象
        Scan scan = new Scan();

        //创建行键过滤器
        RowFilter rowFilter = new RowFilter(CompareOperator.EQUAL, new BinaryPrefixComparator(Bytes.toBytes("47BE1E866CFC071DB19D5E1C056BE28AE24C16E7")));
        //创建单列值过滤器
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("date"), CompareOperator.EQUAL, new SubstringComparator("20180503"));
        SingleColumnValueFilter singleColumnValueFilter2 = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("city"), CompareOperator.EQUAL, new SubstringComparator("83211"));



        //创建一个FilterList对象
        FilterList filterList = new FilterList();
        filterList.addFilter(rowFilter);
        filterList.addFilter(singleColumnValueFilter);
        filterList.addFilter(singleColumnValueFilter2);

        scan.setFilter(filterList);

        ResultScanner rs = table.getScanner(scan);

        Result result = null;

        while ((result=rs.next())!=null){
            StringBuilder sb = new StringBuilder();
            List<Cell> cells = result.listCells();
            String id = Bytes.toString(result.getRow());
            String[] strings = id.split("_");
            String phoneNum = strings[0];
            sb.append("手机号：").append(phoneNum).append("\t");
            long startTime = 20180504000000L-Long.parseLong(strings[1]);
            sb.append("开始时间：").append(startTime).append("\t");
            for (Cell cell : cells) {
                String colNume = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                sb.append(colNume).append("：").append(value).append("\t");
            }

            System.out.println(sb);
        }



    }

    public static void getData(TableName tableName) throws Exception{
        //获取表的实例
        Table table = conn.getTable(tableName);

        //创建Scan的对象
        Scan scan = new Scan();

        RowFilter rowFilter = new RowFilter(CompareOperator.EQUAL, new BinaryPrefixComparator(Bytes.toBytes("47BE1E866CFC071DB19D5E1C056BE28AE24C16E7")));

        scan.setFilter(rowFilter);

        scan.setLimit(10);

        ResultScanner rs = table.getScanner(scan);

        Result result = null;

        while ((result=rs.next())!=null){
            StringBuilder sb = new StringBuilder();
            List<Cell> cells = result.listCells();
            String id = Bytes.toString(result.getRow());
            String[] strings = id.split("_");
            String phoneNum = strings[0];
            sb.append("手机号：").append(phoneNum).append("\t");
            long startTime = 20180504000000L-Long.parseLong(strings[1]);
            sb.append("开始时间：").append(startTime).append("\t");
            for (Cell cell : cells) {
                String colNume = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                sb.append(colNume).append("：").append(value).append("\t");
            }

            System.out.println(sb);
        }

    }

    public static void putDatas(TableName tableName) throws Exception{
        //获取表
        Table table = conn.getTable(tableName);

        //创建一个List集合
        ArrayList<Put> puts = new ArrayList<>();

        BufferedReader br = new BufferedReader(new FileReader("hbase/data/dianxin_data"));
        String line = null;
        while ((line=br.readLine())!=null){
            String[] strings = line.split("\t");
            if(strings.length>7 && !("\\N".equals(strings[1]))){
                //将手机号和进入网格的时间一起作为rowkey进行设计
                //为了获取最新的时间数据，结合hbaserowkey是字典升序排序的规则
                //将进入网格的时间，进行时间戳反转设计，再与手机号进行拼接
                //20180504000000
                //20180503085933
                long maxNumber = 20180504000000L;
                String phoneNum = strings[0];
                long startTime = Long.parseLong(strings[5]);
                String rowKey = phoneNum+"_"+(maxNumber-startTime);

                //创建一个Put对象
                Put put = new Put(Bytes.toBytes(rowKey));

                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("wg"),Bytes.toBytes(strings[1]));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("city"),Bytes.toBytes(strings[2]));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("qx"),Bytes.toBytes(strings[3]));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("stayTime"),Bytes.toBytes(strings[4]));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("stopTime"),Bytes.toBytes(strings[6]));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("date"),Bytes.toBytes(strings[7]));

                puts.add(put);
            }
        }

        table.put(puts);

    }

    public static void createTable(TableName tableName) throws Exception{

        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);

        //创建列簇描述器
        ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of("info");

        tableDescriptorBuilder.setColumnFamily(info);

        admin.createTable(tableDescriptorBuilder.build());

        System.out.println("表创建成功！");
    }

    public static void init() throws Exception{
        //获取配置文件对象
        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        //获取数据库连接
        conn = ConnectionFactory.createConnection(conf);

        //获取数据库操作对象
        admin = conn.getAdmin();

        System.out.println("连接获取成功："+admin+"| "+conn);


    }
}
